package org.spring.mqtt;

import com.alibaba.fastjson2.JSON;
import lombok.Getter;
import lombok.extern.slf4j.Slf4j;
import org.eclipse.paho.client.mqttv3.*;
import com.yst.spring.context.SpringContext;

import java.util.Hashtable;
import java.util.LinkedList;
import java.util.List;
import java.util.Objects;
import java.util.concurrent.ScheduledThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;

@Slf4j
public class MQTTUtils {

    public interface IOnMessage{
        void onMessage(byte[] data);
    }

    public static final String PUBLIC_TOPIC="public";

    private final Hashtable<String, LinkedList<IOnMessage>> listHashtable=new Hashtable<>();

    @Getter
    private MqttClient mqttClient=null;

    private volatile boolean isConnected=false;

    private void doOnMessage(String topic,byte[] data){
        LinkedList<IOnMessage> onMessages=listHashtable.get(topic);
        if (Objects.nonNull(onMessages)){
            onMessages.forEach((a)->{
                a.onMessage(data);
            });
        }
    }

    private final List<String> topicList=new LinkedList<>();

    private final MqttCallback callback=new MqttCallback() {
        @Override
        public void connectionLost(Throwable throwable) {
            log.error("mqtt connect lost", throwable);
            isConnected=false;
            startCheckConnect();
        }

        @Override
        public void messageArrived(String s, MqttMessage mqttMessage) throws Exception {
            if (Objects.nonNull(mqttMessage)&&Objects.nonNull(mqttMessage.getPayload())) {
                doOnMessage(s, mqttMessage.getPayload());
            }
            else {
                doOnMessage(s,null);
            }
            isConnected=true;
        }

        @Override
        public void deliveryComplete(IMqttDeliveryToken iMqttDeliveryToken) {

        }
    };

    public void on(String topic,IOnMessage callback){
        LinkedList<IOnMessage> list = listHashtable.computeIfAbsent(topic, (map) -> new LinkedList<>());
        if (!list.contains(callback)) {
            list.add(callback);
        }
        if (Objects.nonNull(mqttClient)){
            try {
                mqttClient.subscribe(topic);
            }
            catch (Exception e){
                log.error(e.getMessage(),e);
            }
        }
        if (!topicList.contains(topic)) {
            topicList.add(topic);
        }
    }

    public void publish(String topic,Object content) throws MqttException {
        if (Objects.nonNull(mqttClient)&&mqttClient.isConnected()) {
            byte[] data = JSON.toJSONBytes(content);
            MqttMessage mqttMessage = new MqttMessage();
            mqttMessage.setPayload(data);
            mqttClient.publish(topic, mqttMessage);
        }
        else {
            throw new RuntimeException("mqtt not ready!");
        }
    }

    public void publish(Object content) throws MqttException {
        publish(PUBLIC_TOPIC,content);
    }

    private void batchSubscribe() throws MqttException {
        mqttClient.subscribe(topicList.toArray(new String[0]));
    }

    private final AtomicBoolean inConnecting=new AtomicBoolean(false);
    private synchronized void connect() {
        if (!inConnecting.getAndSet(true)) {
            try {
                MqttConnectOptions options = SpringContext.getContext()
                        .getBean(MqttConnectOptions.class);
                options.setAutomaticReconnect(true);
                options.setKeepAliveInterval(2000);
                options.setCleanSession(false);
                log.info("mqtt connect start!");
                if (Objects.nonNull(mqttClient.getServerURI())) {
                    log.info(String.format("mqtt server:%s", mqttClient.getServerURI()));
                }
                mqttClient.setCallback(callback);
                mqttClient.connect(options);
                isConnected = true;
                log.info("mqtt connect success!");
            }
            catch (Exception e){
                log.error(e.getMessage(),e);
            }
            finally {
                inConnecting.set(false);
            }
        }
    }

    private final SpringContext.IContextReady onReady=(context)->{
        mqttClient=context.getBean(MqttClient.class);
        try {
            connect();
            batchSubscribe();
        } catch (MqttException e) {
            log.error("mqtt connect error!",e);
            throw new RuntimeException(e);
        }
    };

    private ScheduledThreadPoolExecutor scheduledThreadPoolExecutor;
    private synchronized void startCheckConnect(){
        if (Objects.isNull(scheduledThreadPoolExecutor)) {
            scheduledThreadPoolExecutor = new ScheduledThreadPoolExecutor(1);
            scheduledThreadPoolExecutor.scheduleAtFixedRate(() -> {
                if (mqttClient.isConnected() && !isConnected) {
                    try {
                        batchSubscribe();
                    } catch (Exception e) {
                        log.error(e.getMessage(), e);
                    }
                }

                if (!mqttClient.isConnected()) {
                    try {
                        connect();
                        batchSubscribe();
                    } catch (Exception e) {
                        log.error(e.getMessage(), e);
                    }
                }
            }, 0, 2, TimeUnit.MINUTES);
        }
    }

    private MQTTUtils(){
        SpringContext.addOnReadyListener(onReady);
    }

    private static MQTTUtils instance;

    public static MQTTUtils getInstance(){
        if (Objects.nonNull(instance)) return instance;
        synchronized (MQTTUtils.class){
            if (Objects.isNull(instance)) {
                instance = new MQTTUtils();
            }
        }
        return instance;
    }

}
